package com.gtp.kafka.consumer;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import com.gtp.kafka.KafkaConstants;

public abstract class BaseConsumer {
	
	protected KafkaConsumer<String, String> consumer;
	
	public BaseConsumer(){
		consumer=new KafkaConsumer<>(baseConfigs());
	}
		
	private Map<String,Object> baseConfigs() {
		
		Map<String,Object> props=new HashMap<String,Object>();
		props.put("bootstrap.servers", KafkaConstants.servers);
		
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		
		//允许子类自定义
		customConfigs(props);
		
		//group id
		props.put("group.id", "test1");//组
		props.put("isolation.level", "read_committed");//读已提交
		
		//client id
		props.put("client.id","test-consumer-001");
		
		return props;
	}
	
	protected void customConfigs(Map<String,Object> configs){
		//子类可选择性扩展
	}
}
